package guardian

import (
	"context"
	"errors"
	"fmt"
	"github.com/Unixeno/islands/island"
	"github.com/Unixeno/islands/island/log"
	"github.com/Unixeno/islands/island/proxy"
	"github.com/Unixeno/islands/lode"
	"github.com/Unixeno/islands/pb"
	"github.com/Unixeno/islands/stringid"
	"github.com/containerd/console"
	"github.com/golang/protobuf/ptypes/empty"
	"net"
	"strings"
	"sync"
	"time"
)

func (s *server) ListContainer(ctx context.Context, req *empty.Empty) (*pb.ListContainerResp, error) {
	containers := make([]*pb.ContainerInfo, 0)
	for _, container := range island.List() {
		image, _ := lode.OpenImage(container.ImageId)

		containers = append(containers, &pb.ContainerInfo{
			Id:         container.Id,
			Name:       container.Name,
			Image:      image.GetName(),
			ImageId:    container.ImageId,
			Command:    container.Command,
			Created:    container.Created.String(),
			SizeRw:     0,
			SizeRootFs: 0,
			State:      container.Status,
		})
	}
	return &pb.ListContainerResp{
		Containers: containers,
	}, nil
}

func (s *server) CreateContainer(ctx context.Context, req *pb.CreateContainerReq) (*pb.CreateContainerResp, error) {
	logger.Debug("create container req ", req)
	if err := stringid.ValidateID(req.ImageId); err != nil {
		return &pb.CreateContainerResp{Err: newErr(1, err)}, err
	}
	image, err := lode.OpenImage(req.ImageId)
	if err != nil {
		return &pb.CreateContainerResp{Err: newErr(1, err)}, err
	}

	image.Register()
	factory, err := island.Create(req.ImageId, image.Config())
	if err != nil {
		return &pb.CreateContainerResp{Err: newErr(1, err)}, err
	}
	if err := factory.SetName(req.Name); err != nil {
		return &pb.CreateContainerResp{Err: newErr(1, err)}, err
	}
	factory.SetImage(image.MountPoint())

	factory.SetTTY(req.Tty)
	factory.SetCmd(req.Cmd)
	factory.SetEntrypoint(req.Entrypoint)
	factory.SetEnv(req.Env)
	factory.SetWorkingDir(req.WorkingDir)

	if req.Resources != nil {
		resources, err := getContainerResources(req.Resources)
		if err != nil {
			return &pb.CreateContainerResp{Err: newErr(1, err)}, err
		}
		factory.SetCgroups(island.CgroupsConfig{
			CgroupsPath:    req.Resources.CgroupParent,
			LinuxResources: resources,
		})
	}

	if err := factory.SetNamespace(getContainerNamespaces(req.Namespaces)); err != nil {
		return &pb.CreateContainerResp{Err: newErr(1, err)}, err
	}

	if req.Network != nil {
		factory.SetHostname(req.Network.Hostname)
		factory.SetHosts(convertHostsFromPB(req.Network.ExtraHosts))
		factory.SetDNS(req.Network.Dns)

		containerInterface := make([]*island.Network, len(req.Network.Interfaces))
		for index, i := range req.Network.Interfaces {
			address := make([]string, len(i.Address))
			for index, addr := range i.Address {
				address[index] = fmt.Sprintf("%s/%d", addr.Ip, addr.Mask)
			}
			containerInterface[index] = &island.Network{
				Name:       i.Name,
				Bridge:     i.Bridge,
				MacAddress: i.Mac,
				Address:    address,
				Gateway:    i.Gateway,
			}
		}
		factory.SetNetworks(containerInterface)
	}

	if req.Volumes != nil {
		volumes := make([]*island.ContainerVolume, len(req.Volumes))
		for index, v := range req.Volumes {
			flag := island.ReadPermission
			if !v.Readonly {
				flag |= island.WritePermission
			}
			volumes[index] = &island.ContainerVolume{
				Destination: v.Destination,
				RW:          flag,
				Source:      v.Source,
				Name:        v.Name,
			}
		}
		factory.SetVolumes(volumes)
	}

	if req.ExposedPorts != nil {
		proxies := make([]proxy.ProxyInfo, 0)
		for dst, portBinds := range req.ExposedPorts {
			// judge dst port[/tcp | /udp]
			tmp := strings.Split(dst, "/")
			var dstPort, protocol string
			allProtocol := false
			if len(tmp) == 1 {
				if !proxy.IsPort(tmp[0]) {
					return &pb.CreateContainerResp{
						Err: newErr(1, fmt.Errorf("error exposed-ports")),
					}, fmt.Errorf("error exposed-ports")
				}
				allProtocol = true
				dstPort = tmp[0]
			} else if len(tmp) == 2 {
				if !proxy.IsPort(tmp[0]) || !(tmp[1] == "tcp" || tmp[1] == "udp") {
					return &pb.CreateContainerResp{
						Err: newErr(1, fmt.Errorf("error exposed-ports")),
					}, fmt.Errorf("error exposed-ports")
				}
				dstPort = tmp[0]
				protocol = tmp[1]
			} else {
				return &pb.CreateContainerResp{
					Err: newErr(1, fmt.Errorf("error exposed-ports")),
				}, fmt.Errorf("error exposed-ports")
			}

			for _, portBind := range portBinds.PortBindings {
				if !proxy.IsPort(portBind.HostPort) || !proxy.IsSrcIP(portBind.HostIp) {
					return &pb.CreateContainerResp{
						Err: newErr(1, fmt.Errorf("error exposed-ports")),
					}, fmt.Errorf("error exposed-ports")
				}
				if !allProtocol {
					proxies = append(proxies, proxy.ProxyInfo{
						Src:      portBind.HostIp + ":" + portBind.HostPort,
						DstPort:  dstPort,
						Protocol: protocol,
					})
				} else {
					proxies = append(proxies, proxy.ProxyInfo{
						Src:      portBind.HostIp + ":" + portBind.HostPort,
						DstPort:  dstPort,
						Protocol: "udp",
					})
					proxies = append(proxies, proxy.ProxyInfo{
						Src:      portBind.HostIp + ":" + portBind.HostPort,
						DstPort:  dstPort,
						Protocol: "tcp",
					})
				}
			}
		}
		factory.SetExposedPorts(proxies)
	}

	if container, err := factory.Save(); err != nil {
		return &pb.CreateContainerResp{Err: newErr(1, err)}, err
	} else {
		return &pb.CreateContainerResp{Id: container.GetID()}, nil
	}
}

func (s *server) InspectContainer(ctx context.Context, req *pb.ContainerID) (*pb.InspectContainerResp, error) {
	if container, err := island.Open(req.Id); err == nil {
		info := container.Inspect()

		containerState := &pb.ContainerState{
			State:      info.State.String(),
			Pid:        int32(info.Pid),
			ExitCode:   int32(info.ExitCode),
			StartTime:  info.StartTime.Format(time.RFC3339),
			FinishTime: info.FinishTime.Format(time.RFC3339),
		}

		image, _ := lode.OpenImage(container.GetImageID())
		env := make([]string, len(info.RuntimeSetting.Env))
		for key, val := range info.RuntimeSetting.Env {
			env = append(env, key+"="+val)
		}
		containerConfig := &pb.ContainerConfig{
			ImageName:  image.GetName(),
			Env:        env,
			Tty:        info.RuntimeSetting.TTY,
			Entrypoint: info.RuntimeSetting.Entrypoint,
			Cmd:        info.RuntimeSetting.Cmd,
			WorkingDir: info.RuntimeSetting.WorkingDir,
		}

		hostConfig := &pb.HostConfig{}
		if info.CgroupSetting != nil {
			hostConfig.Cgroup = info.CgroupSetting.CgroupsPath
			resources := info.CgroupSetting.LinuxResources
			if resources != nil {
				if resources.CPU != nil && resources.CPU.Shares != nil {
					hostConfig.CPUShare = int32(*resources.CPU.Shares)
				}
				if resources.Memory != nil && resources.Memory.Limit != nil {
					hostConfig.Memory = *resources.Memory.Limit
				}
			}
		}

		interfaces := make([]*pb.NetworkInterface, len(info.RuntimeSetting.Networks))
		for i, n := range info.RuntimeSetting.Networks {
			address := make([]*pb.NetworkAddress, len(n.Address))

			for j, a := range n.Address {
				ip, ipNet, err := net.ParseCIDR(a)
				if err != nil {
					return &pb.InspectContainerResp{Info: nil, Err: newErr(1, err)}, err
				}
				strIp := ip.String()
				mask, _ := ipNet.Mask.Size()
				address[j] = &pb.NetworkAddress{
					Ip:   strIp,
					Mask: int32(mask),
				}
			}
			interfaces[i] = &pb.NetworkInterface{
				Name:    n.Name,
				Bridge:  n.Bridge,
				Gateway: n.Gateway,
				Address: address,
			}
		}

		networkConfig := &pb.Network{
			Hostname:   info.InitSetting.Hostname,
			ExtraHosts: convertHostsToPB(info.InitSetting.Hosts),
			Dns:        info.InitSetting.DNS,
			Interfaces: interfaces,
		}

		execInstances := []*pb.ExecInstance{}

		for _, exec := range info.ExecInstances {
			execInstances = append(execInstances, &pb.ExecInstance{
				ExecId:      exec.ExecId,
				ContainerId: exec.ContainerId,
				Stdin:       exec.Stdin,
				Stdout:      exec.Stdout,
				Stderr:      exec.Stderr,
				Running:     exec.Running,
			})
		}

		containerInfo := &pb.Container{
			Id:            container.GetID(),
			Name:          container.GetName(),
			ImageId:       container.GetImageID(),
			Created:       info.State.CreateTime.Format(time.RFC3339),
			RestartCount:  0,
			LogPath:       "",
			SizeRw:        0,
			SizeRootFs:    0,
			HostConfig:    hostConfig,
			State:         containerState,
			Config:        containerConfig,
			Network:       networkConfig,
			ExecInstances: execInstances,
		}
		return &pb.InspectContainerResp{
			Info: containerInfo,
			Err:  nil,
		}, nil
	} else {
		return &pb.InspectContainerResp{Info: nil, Err: newErr(1, err)}, err
	}
}

func (s *server) StartContainer(ctx context.Context, req *pb.ContainerID) (*pb.Error, error) {
	logger.WithField("id", req.Id).Debug("starting container")
	if err := stringid.ValidateID(req.Id); err != nil {
		return newErr(1, err), err
	}

	container, err := island.Open(req.Id)
	if err != nil {
		return newErr(1, err), err
	}
	image, err := lode.OpenImage(container.GetImageID())
	if err != nil {
		return newErr(1, err), err
	}

	if _, err := image.Mount(); err != nil {
		return newErr(1, err), nil
	}
	defer func() {
		if err := image.Unmount(); err != nil {
			logger.WithError(err).WithField("mount_point", image.MountPoint()).Error("can't unmount image after start container")
		}
	}()
	if err := container.Start(); err != nil {
		return newErr(1, err), err
	}
	return &pb.Error{}, nil
}

func (s *server) StopContainer(ctx context.Context, req *pb.StopContainerReq) (*pb.Error, error) {
	logger.WithField("id", req.Id).WithField("timeout", req.Timeout).Debug("request stop container")
	container, err := island.Open(req.Id)
	if err != nil {
		logger.WithField("id", req.Id).Error("container open failed")
		return newErr(7, err), err
	}
	if !container.IsRunning() {
		return &pb.Error{}, nil
	}

	if err := container.Stop(int(req.Timeout)); err != nil {
		return newErr(7, err), err
	}

	return &pb.Error{}, nil
}

func (s *server) KillContainer(ctx context.Context, req *pb.KillContainerReq) (*pb.Error, error) {
	logger.WithField("signal", req.Signal).Debug("request kill container")
	if err := stringid.ValidateID(req.Id); err != nil {
		return newErr(1, err), err
	}

	container, err := island.Open(req.Id)
	if err != nil {
		return newErr(1, err), err
	}

	if err := container.Kill(req.Signal, true); err != nil {
		return newErr(1, err), err
	}
	return &pb.Error{}, nil
}

func (s *server) RestartContainer(ctx context.Context, req *pb.RestartContainerReq) (*pb.Error, error) {
	logger.WithField("id", req.Id).WithField("timeout", req.Timeout).Debug("request restart container")
	container, err := island.Open(req.Id)
	if err != nil {
		logger.WithField("id", req.Id).Error("container open failed")
		return newErr(1, err), err
	}

	image, err := lode.OpenImage(container.GetImageID())
	if err != nil {
		return newErr(1, err), err
	}

	if _, err := image.Mount(); err != nil {
		return newErr(1, err), nil
	}
	defer func() {
		if err := image.Unmount(); err != nil {
			logger.WithError(err).WithField("mount_point", image.MountPoint()).Error("can't unmount image before start container")
		}
	}()

	if err := container.Restart(int(req.Timeout)); err != nil {
		logger.WithError(err).Error("failed to restart container")
		return newErr(1, err), err
	}

	return &pb.Error{}, nil
}

func (s *server) RemoveContainer(ctx context.Context, req *pb.ContainerID) (*pb.Error, error) {
	logger.WithField("container_id", req.Id).Debug("request to remove container")
	container, err := island.Open(req.Id)
	if err != nil {
		return newErr(1, err), err
	}
	image, err := lode.OpenImage(container.GetImageID())
	if err != nil {
		logger.WithError(err).Fatal("failed to open image while remove container")
	}
	if err := island.Remove(req.Id); err != nil {
		logger.WithError(err).Error("failed to remove container")
		return newErr(1, err), err
	} else {
		image.UnRegister()
	}
	return &pb.Error{}, nil
}

func (s *server) RenameContainer(ctx context.Context, req *pb.RenameContainerReq) (*pb.Error, error) {
	if factory, err := island.Edit(req.Id); err != nil {
		logger.WithError(err).Error("failed to rename container")
		return newErr(1, err), err
	} else {
		factory.SetName(req.Name)
		if _, err := factory.Save(); err != nil {
			logger.WithError(err).Fatal("failed to rename container")
		}
	}
	return &pb.Error{}, nil
}

func (s *server) PauseContainer(ctx context.Context, req *pb.ContainerID) (*pb.Error, error) {
	logger.WithField("container_id", req.Id).Debug("request to pause container")
	container, err := island.Open(req.Id)
	if err != nil {
		return newErr(1, err), err
	}
	if err := container.Pause(); err != nil {
		logger.WithError(err).Error("failed to pause container")
		return newErr(1, err), err
	}
	return &pb.Error{}, nil
}

func (s *server) WaitContainer(ctx context.Context, req *pb.WaitContainerReq) (*pb.WaitContainerResp, error) {
	logger.WithField("container_id", req.Id).Debug("request to wait container to stop")
	container, err := island.Open(req.Id)
	if err != nil {
		return &pb.WaitContainerResp{
			ExitCode: -1,
			Err:      newErr(1, err),
		}, err
	}

	var waitChan chan struct{}
	if req.Condition == "not-running" || req.Condition == "" {
		if !container.IsRunning() {
			return &pb.WaitContainerResp{
				ExitCode: int32(container.GetExitCode()),
				Err:      newErr(1, fmt.Errorf("container is not running")),
			}, nil
		}
		waitChan = container.WaitStop()
	} else if req.Condition == "removed" {
		waitChan = container.WaitRemove()
	} else {
		return &pb.WaitContainerResp{
			ExitCode: -1,
			Err:      newErr(1, fmt.Errorf("invalid condition")),
		}, nil
	}

	select {
	case <-waitChan:
		return &pb.WaitContainerResp{
			ExitCode: int32(container.GetExitCode()),
			Err:      nil,
		}, nil
	case <-ctx.Done():
		return &pb.WaitContainerResp{
			ExitCode: -1,
			Err:      newErr(1, fmt.Errorf("client close")),
		}, nil
	}

}

func (s *server) UnpauseContainer(ctx context.Context, req *pb.ContainerID) (*pb.Error, error) {
	logger.WithField("container_id", req.Id).Debug("request to unpause container")
	container, err := island.Open(req.Id)
	if err != nil {
		return newErr(1, err), err
	}
	if err := container.Unpause(); err != nil {
		logger.WithError(err).Error("failed to pause container")
		return newErr(1, err), err
	}
	return &pb.Error{}, nil
}

func (s *server) ResizeContainerTTY(ctx context.Context, req *pb.ResizeTTYReq) (*pb.Error, error) {
	if container, err := island.Open(req.Id); err != nil {
		logger.WithError(err).Error("failed to open a container to resize tty")
	} else {

		if err := container.ResizeTTY(uint16(req.Width), uint16(req.Height)); err != nil {
			logger.WithError(err).Fatal("failed to resize tty")
		}
	}
	return &pb.Error{}, nil
}

func (s *server) CreateExecInstance(ctx context.Context, req *pb.CreateExecInstanceReq) (*pb.ExecID, error) {
	logger.WithField("req", req).Debug("create exec instance")
	options := island.ProcessOptions{
		ContainerId: req.Id,
		Stdin:       true,
		Stdout:      true,
		Stderr:      true,
		Terminal:    req.Tty,
		User:        "root",
		Privileged:  false,
		WorkDir:     req.WorkingDir,
		Command:     req.Cmd,
		Env:         req.Env,
	}

	execid, err := island.CreateExecInstance(options)
	if err != nil {
		return nil, err
	} else {
		return &pb.ExecID{
			Id: execid,
		}, nil
	}

}

func (s *server) StartExecInstance(ctx context.Context, req *pb.ExecID) (*pb.Error, error) {
	logger.WithField("id", req.Id).Debug("start exec instance")
	if exec, err := island.GetExecInstance(req.Id); err != nil {
		logger.WithField("id", req.Id).Error(err)
		return &pb.Error{
			Code:    6,
			Message: err.Error(),
		}, err
	} else {
		if err := exec.StartExecInstance(); err != nil {
			logger.WithField("id", req.Id).Error(err)
			return &pb.Error{
				Code:    6,
				Message: err.Error(),
			}, err
		}
	}
	return &pb.Error{}, nil
}

func (s *server) ResizeExecTTY(ctx context.Context, req *pb.ResizeTTYReq) (*pb.Error, error) {
	logger.WithField("id", req.Id).WithField("height", req.Height).WithField("width", req.Width).Debug("resize exec console")

	exec, err := island.GetExecInstance(req.Id)
	if err != nil {
		logger.WithField("id", req.Id).WithField("error", err.Error()).Error("cannot get exec instance")
		return &pb.Error{
			Code:    6,
			Message: err.Error(),
		}, err
	}

	exec.ResizeExecConsole(console.WinSize{
		Height: uint16(req.Height),
		Width:  uint16(req.Width),
	})

	return &pb.Error{}, nil
}

func (s *server) InspectExec(ctx context.Context, req *pb.ExecID) (*pb.InspectExecResp, error) {
	logger.WithField("id", req.Id).Debug("inspect exec instance")
	execInfo := island.GetExecInfo()
	execInfo.ExecLock.RLock()
	if exec, ok := execInfo.ExecInstances[req.Id]; ok {

		return &pb.InspectExecResp{
			Id:          req.Id,
			Pid:         int32(exec.Process.Pid),
			Tty:         exec.Process.Terminal,
			Env:         exec.Process.Env,
			WorkingDir:  exec.Process.Cwd,
			ContainerId: exec.ContainerId,
			Running:     exec.Running,
			ExitCode:    int32(exec.Process.ExitCode),
		}, nil
	} else {
		logger.WithField("id", req.Id).Error("exec id not existed")
		return &pb.InspectExecResp{}, errors.New("exec id not existed")
	}
}

func (s *server) AttachExec(req pb.Islands_AttachExecServer) error {
	r, err := req.Recv()
	if err != nil {
		logger.WithError(err).Debug("abort attach to exec")
		return err
	}
	logger.WithField("id", r.Id).Debug("attach to exec instance")

	if exec, err := island.GetExecInstance(r.Id); err != nil {
		logger.WithField("id", r.Id).Error("cannot get exec instance")
		return err
	} else {

		defer func() {
			exec.SetExecRunningState(false)
			execInfo := island.GetExecInfo()

			proc := island.GetProcess(exec.Process.Pid)
			exitCode := proc.Wait()

			logger.WithField("exit code", exitCode).Debug("get exit code")

			execInfo.ExecLock.Lock()
			execInfo.ExecInstances[exec.ExecId] = exec
			execInfo.ExecLock.Unlock()
		}()

		grpcIO := island.NewGrpcIO(r.Id, req)
		exec.SetGrpcIO(grpcIO)

		exec.Process.StdinChan = make(chan struct{}, 1)
		exec.Process.StdoutChan = make(chan struct{}, 1)
		exec.Process.StderrChan = make(chan struct{}, 1)

		for {
			select {
			case <-exec.Process.StdinChan:
				return nil
			case <-exec.Process.StdoutChan:
				return nil
			}
		}

	}
}

func (s *server) AttachContainer(req pb.Islands_AttachContainerServer) error {
	r, err := req.Recv()
	if err != nil {
		logger.WithError(err).Debug("abort attach")
		return err
	}
	logger.WithField("id", r.Id).Debug("attach to container")

	if container, err := island.Open(r.Id); err != nil {
		logger.WithError(err).WithField("id", r.Id).Error("can't open container")
		return err
	} else {
		stdin, stdout, _ := container.GetStdio()
		exitChan := make(chan bool, 2)
		go attachStdinHandle(r.Content, req, stdin, exitChan)
		go attachStdoutHandle(req, stdout, exitChan)
		<-exitChan
	}
	return nil
}

func attachStdinHandle(content []byte, req pb.Islands_AttachContainerServer, stdin chan []byte, exit chan bool) {
	remoteStdin := make(chan []byte, 1)
	if content != nil {
		remoteStdin <- content
	}

	go func() {
		for {
			if r, err := req.Recv(); err != nil {
				logger.WithError(err).Debug("stop")
				close(remoteStdin)
				return
			} else {
				remoteStdin <- r.Content
			}
		}
	}()
	for chunk := range remoteStdin {
		stdin <- chunk
	}
	close(stdin)
	exit <- true
}

func attachStdoutHandle(req pb.Islands_AttachContainerServer, stdout chan []byte, exit chan bool) {
	for buffer := range stdout {
		if err := req.Send(&pb.AttachStreamOut{Content: buffer}); err == nil {
			continue
		} else {
			// 可能这里需要丢弃chan内数据避免island阻塞
			logger.WithError(err).Debug("error while sending")
			return
		}
	}
	exit <- true
}

func convertHostsFromPB(hosts []*pb.HostEntry) []*island.HostEntry {
	islandHosts := make([]*island.HostEntry, len(hosts))
	for index, entry := range hosts {
		islandHosts[index] = &island.HostEntry{
			Host:    entry.Host,
			Address: entry.Ip,
		}
	}
	return islandHosts
}

func convertHostsToPB(hosts []*island.HostEntry) []*pb.HostEntry {
	pbHosts := make([]*pb.HostEntry, len(hosts))
	for index, host := range hosts {
		pbHosts[index] = &pb.HostEntry{Ip: host.Address, Host: host.Host}
	}
	return pbHosts
}

func (s *server) CommitContainer(ctx context.Context, req *pb.CommitContainerReq) (*pb.CommitContainerResp, error) {
	if container, err := island.Open(req.Id); err != nil {
		logger.WithError(err).Error("failed to get container")
		return &pb.CommitContainerResp{Err: newErr(1, err)}, err
	} else {
		id, err := lode.Commit(container.GetImageID(), container.GetRWLayer(), req.Name)
		if err != nil {
			logger.WithError(err).Error("failed to commit")
			return &pb.CommitContainerResp{Err: newErr(1, err)}, err
		}
		return &pb.CommitContainerResp{
			Id:  id,
			Err: &pb.Error{},
		}, nil
	}
}

func (s *server) FetchLog(req *pb.FetchLogReq, stream pb.Islands_FetchLogServer) error {
	logger.WithField("req", req).Debug("Fetch log")

	var logStreamer *log.StreamLogger
	if req.Container {
		if container, err := island.Open(req.Id); err != nil {
			logger.WithField("id", req.Id).Error("cannot open container")
			return err
		} else {
			logStreamer = container.GetLogs()
		}
	} else {
		if exec, err := island.GetExecInstance(req.Id); err != nil {
			logger.WithField("id", req.Id).Error("cannot get exec instance")
			return err
		} else {
			logStreamer = exec.GetLogs()
		}
	}

	config := log.ReadConfig{
		Since:  req.Since,
		Tail:   int(req.Tail),
		Follow: req.Follow,
	}

	streamSend := func(stdioName string, msg *log.Entry) error {
		var resp *pb.FetchLogResp
		entry := []*pb.LogEntry{
			{
				Time: msg.LogTime,
				Line: msg.Line,
			},
		}

		switch stdioName {
		case "stdin":
			resp = &pb.FetchLogResp{
				LogStdinEntry: entry,
			}
		case "stdout":
			resp = &pb.FetchLogResp{
				LogStdoutEntry: entry,
			}
		case "stderr":
			resp = &pb.FetchLogResp{
				LogStderrEntry: entry,
			}
		}

		if err := stream.Send(resp); err != nil {
			logger.WithField("error", err).Error("cannot send message")
			return err
		} else {
			return nil
		}
	}

	monitorStream := func(name string, watcher *log.Watcher, group *sync.WaitGroup) {
		defer group.Done()
		for {
			select {
			case msg, ok := <-watcher.Msg:
				if !ok {
					logger.Debug("close log stream")
					watcher.Close()
					return
				}
				if err := streamSend(name, msg); err != nil {
					logger.WithField("error", err).Error("cannot send message")
					return
				}
			case err := <-watcher.Err:
				logger.WithField("error", err).Error("fetch logs failed")
				watcher.Close()
				return
			}
		}
	}

	group := &sync.WaitGroup{}

	if req.Stdin {
		stdinWatcher := logStreamer.Stdin.ReadLogs(config)
		group.Add(1)
		go monitorStream("stdin", stdinWatcher, group)
	}

	if req.Stdout {
		stdoutWatcher := logStreamer.Stdout.ReadLogs(config)
		group.Add(1)
		go monitorStream("stdout", stdoutWatcher, group)
	}

	if req.Stderr {
		stderrWatcher := logStreamer.Stderr.ReadLogs(config)
		group.Add(1)
		go monitorStream("stderr", stderrWatcher, group)
	}

	group.Wait()

	return nil
}
